package thread.disruptor.multi;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 在多生产者多消费者模式下，每个监听（EventHandler）不依赖一个单独的线程，而是委托给线程池；
 *
 * @author verlet
 * @date 2020/5/6
 */
public class Main {
    public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
        RingBuffer<Order> ringBuffer = RingBuffer.create(
                ProducerType.MULTI, new EventFactory<Order>() {
                    @Override
                    public Order newInstance() {
                        return new Order();
                    }
                }, 1024 * 1024, new YieldingWaitStrategy());

        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        Consumer[] consumers = new Consumer[10];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("C" + i);
        }
        WorkerPool<Order> orderWorkerPool = new WorkerPool<Order>(
                ringBuffer, sequenceBarrier, new IntEventExceptionHandler(), consumers);

        // 把消费者的位置信息引用注入到生产者
        ringBuffer.addGatingSequences(orderWorkerPool.getWorkerSequences());

        ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        orderWorkerPool.start(executorService);

        CyclicBarrier barrier = new CyclicBarrier(100);
        for (int i = 0; i < 100; i++) {
            final Producer p = new Producer(ringBuffer);
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        barrier.await();
                    } catch (Exception  e) {
                        e.printStackTrace();
                    }
                    for (int j = 0; j < 100; j++) {
                        p.onData(UUID.randomUUID().toString());
                    }

                }
            }).start();
        }
        System.out.println("---------------开始生产-----------------");
        Thread.sleep(5000);
        System.out.println("总数:" + Consumer.getCount());
        orderWorkerPool.halt();
        executorService.shutdown();
    }

    private static class IntEventExceptionHandler implements ExceptionHandler<Order> {

        // 进行消费的时候出现异常
        @Override
        public void handleEventException(Throwable ex, long sequence, Order event) {

        }

        // 启动时候
        @Override
        public void handleOnStartException(Throwable ex) {

        }

        // 消费结束的时候
        @Override
        public void handleOnShutdownException(Throwable ex) {

        }
    }

}
